package com.qzuser.sendmessagebasic.server.dao;

import com.qzuser.sendmessagebasic.common.exception.MemoryException;
import com.qzuser.sendmessagebasic.common.exception.MessageFileException;
import com.qzuser.sendmessagebasic.server.binding.Binding;
import com.qzuser.sendmessagebasic.server.exchange.Exchange;
import com.qzuser.sendmessagebasic.server.message.Message;
import com.qzuser.sendmessagebasic.server.queue.Queue;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

public class MemoryCenter {
    private final ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Queue> queueMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, LinkedList<Message>> queueMsgMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> waitAckMsgMap = new ConcurrentHashMap<>();

    public void addExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
    }

    public void addQueue(Queue queue) {
        queueMap.put(queue.getName(), queue);
    }

    public Queue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
    }

    public void addBinding(Binding binding) throws MemoryException {
        ConcurrentHashMap<String, Binding> exchangeBind = bindingMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
        synchronized (exchangeBind) {
            if (exchangeBind.get(binding.getQueueName()) != null) {
                throw new MemoryException("[MemoryCenter] 绑定已存在：" + binding.getExchangeName()+ "-" +binding.getQueueName());
            }
            exchangeBind.put(binding.getQueueName(), binding);
        }
    }

    public Binding getBinding(String exchangeName, String queueName) {
        if (bindingMap.get(exchangeName) == null) {
            return null;
        }
        return bindingMap.get(exchangeName).get(queueName);
    }

    public void deleteBinding(String exchangeName, String queueName) throws MemoryException {
        ConcurrentHashMap<String, Binding> exchangeBind = bindingMap.get(exchangeName);
        if (exchangeBind == null) {
            throw new MemoryException("[MemoryCenter] 绑定关系不存在：" + exchangeName + "-" +queueName );
        }
        exchangeBind.remove(queueName);
    }

    public ConcurrentHashMap<String, Binding> getAllBinding(String exchangeName) {
        return bindingMap.get(exchangeName);
    }

    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
    }

    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    public void deleteMessage(String messageId) {
        messageMap.remove(messageId);
    }

    public void sendMessage(Queue queue, Message message) {
        LinkedList<Message> messages = queueMsgMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        synchronized (messages) {
            messages.add(message);
            addMessage(message);
        }
    }

    public Message pushMessage(String queueName) {
        LinkedList<Message> messages = queueMsgMap.get(queueName);
        if (messages == null || messages.isEmpty()) {
            return null;
        }
        synchronized (messages) {
            return messages.remove(0);
        }
    }

    public LinkedList<Message> getMessages(String queueName) {
        return queueMsgMap.get(queueName);
    }

    public int getQueueMsgCount(String queueName) {
        LinkedList<Message> messages = queueMsgMap.get(queueName);
        if (messages == null || messages.isEmpty()) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    public void addWaitAckMsg(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageMap = waitAckMsgMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(), message);
    }

    public void deleteWaitAckMsg(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = waitAckMsgMap.get(queueName);
        if (messageMap == null || messageMap.isEmpty()) {
            return;
        }
        messageMap.remove(messageId);
    }

    public Message getWaitAckMsg(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = waitAckMsgMap.get(queueName);
        if (messageMap == null || messageMap.isEmpty()) {
            return null;
        }
        return messageMap.get(messageId);
    }

    public void recovery(DiskCenter diskCenter) {
        exchangeMap.clear();
        queueMap.clear();
        bindingMap.clear();
        messageMap.clear();
        queueMsgMap.clear();
        diskCenter.getExchanges().forEach(exchange -> {
            exchangeMap.put(exchange.getName(), exchange);
        });
        List<Queue> queues = diskCenter.getQueues();
        queues.forEach(queue -> {
            queueMap.put(queue.getName(), queue);
        });
        diskCenter.getBindings().forEach(binding -> {
            bindingMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>())
                    .put(binding.getQueueName(), binding);
        });
        queues.forEach(queue -> {
            try {
                LinkedList<Message> queueMsg = diskCenter.loadMessages(queue.getName());
                queueMsgMap.put(queue.getName(), queueMsg);
                queueMsg.forEach((message -> {
                    messageMap.put(message.getMessageId(), message);
                }));
            } catch (MessageFileException e) {
                e.printStackTrace();
            }
        });
    }
}
